import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class test1 {
     public static void main(String[] args) {
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
       env.setParallelism(1);

       //需求：通过广播一个标记，让算子的每个并行实例按照标记执行不同的逻辑代码
       //得到两条流：数据流和广播流

       //数据流
       DataStreamSource<String> ds = env.socketTextStream("hadoop102", 9999);

       //广播流
       //基于普通流进行广播操作
       MapStateDescriptor mapStateDesc = new MapStateDescriptor<>("mapStateDesc", Types.STRING, Types.STRING);//todo 这里导包不要导错了，导flink下的type包

       BroadcastStream<String> broadcastDs =
               env.socketTextStream("hadoop102", 8888).broadcast(mapStateDesc);

       //connect
       ds.connect( broadcastDs )
               .process(
                       new BroadcastProcessFunction<String, String, String>() {
                           /*
                           *
                           * 处理数据流中的数据
                           *
                           * 从广播状态中获取状态数据，结合上状态数据，对数据流中的数据进行处理
                           *
                           * 对广播状态进行读取，使用状态
                           * */
                           @Override
                           public void processElement(String value, BroadcastProcessFunction<String, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {

                               //获取广播状态
                               ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDesc);

                               String flag = broadcastState.get("flag"); //todo 这个flag应该是从外部传过去的

                               if ("1".equals(flag)){
                                   System.out.println("执行1号逻辑处理。。。。。");
                               }else if ("2".equals(flag)){
                                   System.out.println("执行2号处理逻辑。。。。。");
                               } else if ("3".equals(flag)) {
                                    System.out.println("执行3号处理逻辑。。。。。");
                               }else {
                                   System.out.println("执行默认处理逻辑。。。。。");
                               }
                               //输出
                               out.collect( value );

                           }
                           /**
                             * 处理广播流中的数据
                             *
                             * 广播流中进来数据，将数据添加到广播状态中。
                             *
                             * 对广播状态进行 新增 、 修改 、 删除
                            *
                            * 维护状态
                             */

                           @Override
                           public void processBroadcastElement(String value, BroadcastProcessFunction<String, String, String>.Context context, Collector<String> out) throws Exception {

                               //获取广播状态
                               BroadcastState<String,String> broadcastState = context.getBroadcastState(mapStateDesc);

                               //将广播的数据添加到广播状态中
                               broadcastState.put("flag",value);
                           }
                       }
               )
               .setParallelism(1)
               .print();


       try {
           env.execute();
       } catch (Exception e) {
           throw new RuntimeException(e);
       }
   }
}
